package com.seventeen.mq.proxy.server.proxy.thread;

import com.seventeen.mq.proxy.common.constant.DelFlagConstants;
import com.seventeen.mq.proxy.common.utils.StringUtils;
import com.seventeen.mq.proxy.server.proxy.ext.BaseExtMqConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 消费者代理资源池
 *
 * @author seventeen
 */
public class ConsumerPool extends Thread {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerPool.class);
    private final Map<Long, BaseExtMqConsumer> consumerMap = new ConcurrentHashMap<>(16);

    private final LinkedBlockingQueue<BaseExtMqConsumer> consumerQueue;
    private volatile boolean toStop = false;

    public ConsumerPool() {
        this.consumerQueue = new LinkedBlockingQueue<>();
    }

    public void pushQueue(BaseExtMqConsumer mqConsumer) {
        consumerQueue.add(mqConsumer);
    }

    public void toStop() {
        LOGGER.info("consumerPool toStop!");
        this.toStop = true;
    }

    @Override
    public void run() {
        while (!toStop) {
            try {
                BaseExtMqConsumer mqConsumer = consumerQueue.poll(3L, TimeUnit.SECONDS);
                if (mqConsumer == null) {
                    continue;
                }
                Long id = mqConsumer.getId();
                BaseExtMqConsumer oldMqConsumer = consumerMap.get(id);
                if (oldMqConsumer != null) {

                    if (DelFlagConstants.DELETED.equals(mqConsumer.getDelFlag())) {
                        oldMqConsumer.destroy();
                        consumerMap.remove(id);
                        LOGGER.info("proxy consumerMap update {}", consumerMap);
                        continue;
                    }

                    if (StringUtils.equalsIgnoreCase(mqConsumer.getVersion(), oldMqConsumer.getVersion())) {
                        continue;
                    } else {
                        oldMqConsumer.destroy();
                        consumerMap.remove(id);
                        LOGGER.info("proxy consumerMap update {}", consumerMap);
                    }

                }

                if (DelFlagConstants.DELETED.equals(mqConsumer.getDelFlag())) {
                    continue;
                }
                mqConsumer.init();
                consumerMap.put(id, mqConsumer);
                LOGGER.info("proxy consumerMap update {}", consumerMap);
            } catch (Throwable e) {
                LOGGER.error("proxy consumerMap update error {}, consumerMap {}", e.getMessage(), consumerMap, e);
            }

        }

        LOGGER.info("toStop is {}", toStop);
        if (toStop) {
            LOGGER.info("consumerQueue clear!");
            consumerQueue.clear();
            for (BaseExtMqConsumer extMqConsumer : consumerMap.values()) {
                extMqConsumer.destroy();
            }
        }
    }
}
